package com.rr.learningdemo.flink;

import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.util.Collector;

import java.util.Collections;

/**
 * @author RR
 * @date 2021/7/1 23:21
 */
//dataStream分流合流
public class StreamSplitTest {
    private static int i = -1;
    public static void main(String[] args) throws Exception {

        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //从txt文件读取数据
        String txtFilePath = "G:\\IDEAProject\\learning-demo\\src\\main\\resources\\WordCount.txt";
        DataStream<String> dataStreamSource = env.addSource(new MyNoParalleSource());
        SplitStream<String> split = dataStreamSource.split(new OutputSelector<String>() {
            @Override
            public Iterable<String> select(String s) {
                //放在两条流里面
                return (i *= -1) == 1 ? Collections.singleton("list1") : Collections.singleton("list2");
            }
        });
        DataStream<String> list1 = split.select("list1");
        DataStream<String> list2 = split.select("list2");
//        list1.print();
//        list2.print();

        //<String,String>对应list1和list2的泛型，当然可以两个不一样泛型的流合并
        ConnectedStreams<String, String> connectedStreams = list1.connect(list2);
        //合流，可以合并多条流，但多条流的泛型必需一致
//        DataStream<String> union = list1.union(list2, list2);
        SingleOutputStreamOperator<String> stringSingleOutputStreamOperator = connectedStreams.flatMap(new CoFlatMapFunction<String, String, String>() {
            //上面的这三个泛型分别是指第一个流，第二个流，合并得到的流的泛型，下面分别实现它们的转换
            @Override
            public void flatMap1(String s, Collector<String> collector) throws Exception {
                collector.collect(s.split("从")[0]);
            }

            @Override
            public void flatMap2(String s, Collector<String> collector) throws Exception {
                collector.collect(s.split("从")[0]);
            }
        });

        stringSingleOutputStreamOperator.print();

        env.execute();

    }
}
